package com.zenitera.bigdata.window;

import com.zenitera.bigdata.bean.WaterSensor;
import com.zenitera.bigdata.util.BigdataUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.ProcessingTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.List;

/**
 * Time - Session Windows
 */
public class Flink01_Window_Time_03 {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);

        env
                .socketTextStream("localhost", 6666)
                .map(line -> {
                    String[] data = line.split(",");
                    return new WaterSensor(
                            String.valueOf(data[0]),
                            Long.valueOf(data[1]),
                            Integer.valueOf(data[2])
                    );

                })
                .keyBy(WaterSensor::getId)
                // 定义一个session窗口: gap是3s
                .window(ProcessingTimeSessionWindows.withGap(Time.seconds(3)))
                .process(new ProcessWindowFunction<WaterSensor, String, String, TimeWindow>() {

                    @Override
                    public void process(String key,
                                        Context ctx,
                                        Iterable<WaterSensor> elements,
                                        Collector<String> out) throws Exception {

                        List<WaterSensor> list = BigdataUtil.toList(elements);

                        String stt = BigdataUtil.toDateTime(ctx.window().getStart());
                        String edt = BigdataUtil.toDateTime(ctx.window().getEnd());


                        out.collect("窗口: " + stt + " " + edt + ", key:" + key + "  " + list);

                    }
                })
                .print();


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }


    }
}
/*
D:\netcat-win32-1.12>nc64.exe -lp 6666
a1,1,3
a1,1,3
u1,2,4
u1,2,4
u1,2,4
u1,2,4
u1,2,4
p1,3,10
p1,3,10
p1,3,10
p1,3,10
-----------------------------
窗口: 2023-03-22 15:04:59 2023-03-22 15:05:04, key:a1  [WaterSensor(id=a1, ts=1, vc=3), WaterSensor(id=a1, ts=1, vc=3)]
窗口: 2023-03-22 15:05:07 2023-03-22 15:05:12, key:u1  [WaterSensor(id=u1, ts=2, vc=4), WaterSensor(id=u1, ts=2, vc=4), WaterSensor(id=u1, ts=2, vc=4), WaterSensor(id=u1, ts=2, vc=4), WaterSensor(id=u1, ts=2, vc=4)]
窗口: 2023-03-22 15:05:16 2023-03-22 15:05:22, key:p1  [WaterSensor(id=p1, ts=3, vc=10), WaterSensor(id=p1, ts=3, vc=10), WaterSensor(id=p1, ts=3, vc=10)]
窗口: 2023-03-22 15:05:23 2023-03-22 15:05:26, key:p1  [WaterSensor(id=p1, ts=3, vc=10)]

Process finished with exit code -1
 */